package data

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.Table
import org.apache.flink.table.descriptors.{Csv, FileSystem, OldCsv, Schema}
import org.apache.flink.types.Row

// 开发时要做的第一件事情
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api._
object WindowSQLApp {
  val TABLE_NAME = "access"
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    /**
     * TODO... 实现滚动窗口
     *  窗口大小是10s
     *  keyBy(user)
     * 1000-9999 :  pk 205  xingxing 6
     * 10000-19999: pk 45
     */

    val input = env.fromElements(
      "1000,pk,Spark,75",
      "2000,pk,Flink,65",
      "2000,xingxing,蜡烛,3",
      "3000,pk,CDH,65",
      "9999,xingxing,皮鞭,3",
      "19999,pk,Hive,45"
    ).map(x => {
      val splits = x.split(",")
      val time = splits(0).toLong
      val user = splits(1)
      val goods = splits(2)
      val money = splits(3).toDouble
      (time,user,goods,money)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(Long, String, String, Double)](Time.seconds(0)) {
      override def extractTimestamp(element: (Long, String, String, Double)): Long = element._1
    })

    val tableEnvironment = StreamTableEnvironment.create(env)
    tableEnvironment.createTemporaryView(TABLE_NAME, input,'time, 'user, 'goods, 'money, 'ts.rowtime())

    //val resultTable = tableEnvironment.from(TABLE_NAME)
    //resultTable.printSchema()
    //resultTable.toAppendStream[Row].print()

    val resultTable = tableEnvironment.from(TABLE_NAME)
        .window(Over partitionBy('user) orderBy('ts) as 'win)
        .select('user, 'money.sum().over('win))
    resultTable.toRetractStream[Row].print("API...")

    val sql =
      s"""
        |select
        |user, sum(money) over win
        |from $TABLE_NAME
        |window win as (
        | partition by user
        | order by ts
        |)
        |""".stripMargin
    tableEnvironment.sqlQuery(sql).toRetractStream[Row].map(_._2).print("SQL...")

    env.execute(getClass.getCanonicalName)
  }

  def session(tableEnvironment:TableEnvironment): Unit = {
    val resultTable = tableEnvironment.from(TABLE_NAME)
      .window(Session withGap 10.seconds on 'ts as 'win)
      .groupBy('user, 'win)
      .select('user, 'money.sum().as("total"), 'win.start(), 'win.end())
    resultTable.toRetractStream[Row].print("API...")


    val sql =
      s"""
         |
         |select
         |user, sum(money) as total,
         |SESSION_START(ts, interval '10' second) as win_start,
         |SESSION_END(ts, interval '10' second) as win_end
         |from $TABLE_NAME
         |group by
         |SESSION(ts, interval '10' second), user
         |""".stripMargin

    tableEnvironment.sqlQuery(sql).toRetractStream[Row].map(_._2).print("SQL...")
  }

  def slide(tableEnvironment:TableEnvironment): Unit = {
    val resultTable = tableEnvironment.from(TABLE_NAME)
      .window(Slide over 10.seconds() every 2.seconds() on 'ts as 'win)
      .groupBy('user, 'win)
      .select('user, 'money.sum().as("total"), 'win.start(), 'win.end())
    resultTable.toRetractStream[Row].print("API...")



    val sql =
      s"""
         |
         |select
         |user, sum(money) as total,
         |HOP_START(ts, interval '2' second, interval '10' second) as win_start,
         |HOP_END(ts, interval '2' second, interval '10' second) as win_end
         |from $TABLE_NAME
         |group by
         |HOP(ts, interval '2' second, interval '10' second), user
         |""".stripMargin

    tableEnvironment.sqlQuery(sql).toRetractStream[Row].map(_._2).print("SQL...")
  }

  def tumble(tableEnvironment:TableEnvironment): Unit = {

    val resultTable = tableEnvironment.from(TABLE_NAME)
      .window(Tumble over 10.seconds() on 'ts as 'win)
      .groupBy('user, 'win)
      .select('user, 'money.sum().as("total"), 'win.start(), 'win.end())
    resultTable.toRetractStream[Row].print("API...")




    /**
     * tumble_start(ts, interval '10' second) as win_start,
     * tumble_end(ts, interval '10' second) as win_end
     *
     * ???
     */
    val sql =
      s"""
         |
         |select
         |user, sum(money) as total
         |from $TABLE_NAME
         |group by
         |tumble(ts, interval '10' second), user
         |""".stripMargin

    tableEnvironment.sqlQuery(sql).toRetractStream[Row].map(_._2).print("SQL...")
  }
}
